package com.nepu.gmall.realtime.app.dwd;

import com.nepu.gmall.realtime.util.KafkaUtils;
import com.nepu.gmall.realtime.util.MysqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import java.time.Duration;

/**
 * 本类处理的是交易域订单预处理表：
 *  构建这个表的原因之后有两个业务需求：订单明细表和取消订单明细表的表结构和数据的来源是相同的，为了做这两个业务的时候减少重复的计算
 *  这里先聚合出一张中间表，之后用这张表做后续的业务。
 *  构建这张表是需要用到：订单信息表，订单明细表，订单与活动关联表，订单与购物券关联表，字典表这五张表。
 *  之后对这五张表进行关联
 *  将数据写入到kafka，这里写入数据的时候用到了撤回流的写入，撤回流写入到kafka的方式，是将撤回的数据标记为null
 *  我们在之前构建kafka的连接器的时候，自定义了序列化和范反序列化的方法的时候，是将null数据给过滤掉了。
 * @author chenshuaijun
 * @create 2023-02-27 8:37
 */
public class DwdTradeOrderPreProcess {

    public static void main(String[] args) throws Exception {
        // TODO 0、加载执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 因为下面对于数据的处理是有大量的join，FlinkSQL默认是将状态永久的进行保存的，为了防止状态的永久的保存，
        // 导致大量的空间占用，这里将状态的TTL设置为10s
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(10));

        // 设置checkpoint的信息：设置checkpoint的间隔是5分钟,并且checkpoint的级别是精确一次性
        /*env.enableCheckpointing(5 * 60000L, CheckpointingMode.EXACTLY_ONCE);
        // 设置checkpoint的超时时间是10分钟
        env.getCheckpointConfig().setCheckpointTimeout(10 * 60000L);
        // 设置外部检查点。可以将检查点的元数据信息定期写入外部系统，这样当job失败时，检查点不会被清除。这样如果job失败，可以从检查点恢复job。
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // 设置checkpoint的重启的策略
        env.setRestartStrategy(RestartStrategies.failureRateRestart(10, Time.of(1L, TimeUnit.DAYS), Time.of(3L, TimeUnit.MINUTES)));
        // 设置两个checkpoint之间的最小的间隔时间
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
        // 设置状态后端: 设置状态后端为内存级别
        env.setStateBackend(new HashMapStateBackend());
        // 设置checkpoint的存储的路径
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/checkpoint");
        // 因为我们的HDFS只有atguigu用户才能够操作，所以要将用户设置为atguigu
        System.setProperty("HADOOP_USER_NAME", "atguigu");*/
        // TODO 1、读取topic_db
        String groupId = "DwdTradeOrderPreProcess";
        tableEnv.executeSql(MysqlUtil.getTopicDBLookUpDDL(groupId));
        // TODO 2、读取订单明细表
        Table orderDetailTable = tableEnv.sqlQuery("" +
                "select  " +
                "    `data`['id'] id,  " +
                "    `data`['order_id'] order_id,  " +
                "    `data`['sku_id'] sku_id,  " +
                "    `data`['sku_name'] sku_name,  " +
                "    `data`['order_price'] order_price,  " +
                "    `data`['sku_num'] sku_num,  " +
                "    `data`['create_time'] create_time,  " +
                "    `data`['source_type'] source_type,  " +
                "    `data`['source_id'] source_id,  " +
                "    `data`['split_total_amount'] split_total_amount,  " +
                "    `data`['split_activity_amount'] split_activity_amount,  " +
                "    `data`['split_coupon_amount'] split_coupon_amount,  " +
                "    proc_time  " +
                "from topic_db   " +
                "where `table`='order_detail' and `type`='insert'");

        tableEnv.createTemporaryView("order_detail",orderDetailTable);
//        tableEnv.toAppendStream(orderDetailTable,Row.class).print();
        // TODO 3、读取订单信息表
        Table orderInfoTable = tableEnv.sqlQuery("" +
                "select " +
                "    `data`['id'] id, " +
                "    `data`['consignee'] consignee, " +
                "    `data`['consignee_tel'] consignee_tel, " +
                "    `data`['total_amount'] total_amount, " +
                "    `data`['order_status'] order_status, " +
                "    `data`['user_id'] user_id, " +
                "    `data`['payment_way'] payment_way, " +
                "    `data`['delivery_address'] delivery_address, " +
                "    `data`['order_comment'] order_comment, " +
                "    `data`['out_trade_no'] out_trade_no, " +
                "    `data`['trade_body'] trade_body, " +
                "    `data`['create_time'] create_time, " +
                "    `data`['operate_time'] operate_time, " +
                "    `data`['expire_time'] expire_time, " +
                "    `data`['process_status'] process_status, " +
                "    `data`['tracking_no'] tracking_no, " +
                "    `data`['parent_order_id'] parent_order_id, " +
                "    `data`['province_id'] province_id, " +
                "    `data`['activity_reduce_amount'] activity_reduce_amount, " +
                "    `data`['coupon_reduce_amount'] coupon_reduce_amount, " +
                "    `data`['original_total_amount'] original_total_amount, " +
                "    `data`['feight_fee'] feight_fee, " +
                "    `data`['feight_fee_reduce'] feight_fee_reduce, " +
                "    `data`['refundable_time'] refundable_time, " +
                "    `type`, " +
                "    `old` " +
                "from topic_db  " +
                "where `table`='order_info' and (`type`='insert' or `type`='update')");

        tableEnv.createTemporaryView("order_info",orderInfoTable);
//        tableEnv.toAppendStream(orderInfoTable,Row.class).print();
        // TODO 4、读取订单与购物券关联表
        Table orderDetailCouponTable = tableEnv.sqlQuery("" +
                "select  " +
                "    `data`['id'] id,  " +
                "    `data`['order_id'] order_id,  " +
                "    `data`['order_detail_id'] order_detail_id,  " +
                "    `data`['coupon_id'] coupon_id,  " +
                "    `data`['coupon_use_id'] coupon_use_id,  " +
                "    `data`['sku_id'] sku_id,  " +
                "    `data`['create_time'] create_time  " +
                "from topic_db   " +
                "where `table`='order_detail_coupon' and `type`='insert'");
        tableEnv.createTemporaryView("order_detail_coupon",orderDetailCouponTable);
//        tableEnv.toAppendStream(orderDetailCouponTable,Row.class).print();
        // TODO 5、读取订单与活动关联表
        Table orderDetailActivityTable = tableEnv.sqlQuery("" +
                "select " +
                "    `data`['id'] id, " +
                "    `data`['order_id'] order_id, " +
                "    `data`['order_detail_id'] order_detail_id, " +
                "    `data`['activity_id'] activity_id, " +
                "    `data`['activity_rule_id'] activity_rule_id, " +
                "    `data`['sku_id'] sku_id, " +
                "    `data`['create_time'] create_time " +
                "from topic_db  " +
                "where `table`='order_detail_activity' and `type`='insert'");
        tableEnv.createTemporaryView("order_detail_activity",orderDetailActivityTable);
//        tableEnv.toAppendStream(orderDetailActivityTable,Row.class).print();
        // TODO 6、读取字典表
        tableEnv.executeSql(MysqlUtil.getBaseDicLookUpDDL());
        // TODO 7、将五张表关联
        Table resultTable = tableEnv.sqlQuery("" +
                "select " +
                "    od.id, " +
                "    od.order_id, " +
                "    od.sku_id, " +
                "    od.sku_name, " +
                "    od.order_price, " +
                "    od.sku_num, " +
                "    od.create_time, " +
                "    od.source_type source_type_id, " +
                "    dic.dic_name source_type_name, " +
                "    od.source_id, " +
                "    od.split_total_amount, " +
                "    od.split_activity_amount, " +
                "    od.split_coupon_amount, " +
                "    oi.consignee, " +
                "    oi.consignee_tel, " +
                "    oi.total_amount, " +
                "    oi.order_status, " +
                "    oi.user_id, " +
                "    oi.payment_way, " +
                "    oi.delivery_address, " +
                "    oi.order_comment, " +
                "    oi.out_trade_no, " +
                "    oi.trade_body, " +
                "    oi.operate_time, " +
                "    oi.expire_time, " +
                "    oi.process_status, " +
                "    oi.tracking_no, " +
                "    oi.parent_order_id, " +
                "    oi.province_id, " +
                "    oi.activity_reduce_amount, " +
                "    oi.coupon_reduce_amount, " +
                "    oi.original_total_amount, " +
                "    oi.feight_fee, " +
                "    oi.feight_fee_reduce, " +
                "    oi.refundable_time, " +
                "    oda.id order_detail_activity_id, " +
                "    oda.activity_id, " +
                "    oda.activity_rule_id, " +
                "    odc.id order_detail_coupon_id, " +
                "    odc.coupon_id, " +
                "    odc.coupon_use_id, " +
                "    oi.`type`, " +
                "    oi.`old`," +
                "    current_row_timestamp() row_op_ts "+
                "from order_detail od join order_info oi on od.order_id = oi.id " +
                "left join order_detail_coupon odc on odc.order_detail_id = od.id " +
                "left join order_detail_activity oda on oda.order_detail_id = od.id " +
                "join base_dic for system_time as of od.proc_time as dic on od.source_type = dic.dic_code");

        tableEnv.createTemporaryView("result_table",resultTable);
//        tableEnv.toAppendStream(resultTable,Row.class).print();
        // TODO 8、使用upsert将数据写入到kakfa
        tableEnv.executeSql("" +
                "CREATE TABLE dwd_trade_order_pre_process ( " +
                "    id string, " +
                "    order_id string, " +
                "    sku_id string, " +
                "    sku_name string, " +
                "    order_price string, " +
                "    sku_num string, " +
                "    create_time string, " +
                "    source_type string, " +
                "    dic_name string, " +
                "    source_id string, " +
                "    split_total_amount string, " +
                "    split_activity_amount string, " +
                "    split_coupon_amount string, " +
                "    consignee string, " +
                "    consignee_tel string, " +
                "    total_amount string, " +
                "    order_status string, " +
                "    user_id string, " +
                "    payment_way string, " +
                "    delivery_address string, " +
                "    order_comment string, " +
                "    out_trade_no string, " +
                "    trade_body string, " +
                "    operate_time string, " +
                "    expire_time string, " +
                "    process_status string, " +
                "    tracking_no string, " +
                "    parent_order_id string, " +
                "    province_id string, " +
                "    activity_reduce_amount string, " +
                "    coupon_reduce_amount string, " +
                "    original_total_amount string, " +
                "    feight_fee string, " +
                "    feight_fee_reduce string, " +
                "    refundable_time string, " +
                "    order_detail_activity_id string, " +
                "    activity_id string, " +
                "    activity_rule_id string, " +
                "    order_detail_coupon_id string, " +
                "    coupon_id string, " +
                "    coupon_use_id string, " +
                "    `type` string, " +
                "    `old` map<string,string>, " +
                "    row_op_ts TIMESTAMP_LTZ(3)," +
                "    PRIMARY KEY (id) NOT ENFORCED " +
                ")" + KafkaUtils.getUpsertKafkaDDL("dwd_trade_order_pre_process"));
        // TODO 9、执行
        tableEnv.executeSql("insert into dwd_trade_order_pre_process select * from result_table");

        env.execute();
    }
}
